/*
 * Copyright (c) China Telecom Cloud Technology Co., Ltd. 2024-2025. All rights reserved.
 */

#include <algorithm>
#include <iostream>
#include <unistd.h>
#include <unordered_set>

#include "Common.h"
#include "Logger.h"
#include "DynamoPublisher.h"

DynamoPublisher::DynamoPublisher() {
    _type       = PUBLISHER_TYPE_DYNAMO;
    _pos        = 0;
    _eventIdCnt = 0;

    int32_t ret = create_nats_connection();
    if (ret != KVC_OK) {
        LOG_ERROR_FMT("connect to nats server failed({})! "
                    "please check nats server status.",
                    ret
                );
    }
    return;
};


/**
 *! @brief 按当前块hashId(前缀哈希)对BlockHashId进行升序排列的比较函数
 * 
 * @param x 
 * @param y 
 * @return bool
 */
bool BlockHashIdAscendByExternalHashId(BlockHashId &x, 
                                    BlockHashId &y) {
    return x.hashId < y.hashId;
}



/**
 *! @brief 按前一块的hashId对BlockHashId进行升序排列的比较函数
 * 
 * @param x 
 * @param y 
 * @return bool
 */
bool BlockHashIdAscendByParentHashId(BlockHashId &x, 
                                    BlockHashId &y) {
    return x.parentHashId < y.parentHashId;
}


// pub struct StoredBlockData {
//     external_hash: u64,
//     local_hash: u64,
// }
void DynamoPublisher::publishOneStoredEvent(int64_t parentHashId,
                                            StoredBlockData *blkInfo,
                                            int32_t &blkInfoNum) {
    // 是否为首块
    bool isFirst = (parentHashId == 0);
    LOG_DEBUG_FMT("[Publish Stored Event] hashIdNum:{}, isFirst: {}",
                  blkInfoNum, isFirst);
    
    publishStoredEvent(_eventIdCnt, 
                    parentHashId,
                    blkInfo, 
                    blkInfoNum, 
                    isFirst
                );
    
    _eventIdCnt++;
    blkInfoNum = 0;

    return;
}



void DynamoPublisher::publishStoredEventBatch(
    std::vector<BlockHashId> &addedHashIds) {
    if (!addedHashIds.empty()) {
        LOG_DEBUG_FMT("addHashId:");
        for (auto it : addedHashIds) {
            LOG_DEBUG_FMT("[Add] hashId:{}, parent:{}, local:{}", it.hashId,
                          it.parentHashId, it.localHashId);
        }
        LOG_DEBUG_FMT("addHashId end");
    }

    StoredBlockData blkInfo[MAX_HASHIDS_PER_NATS_REQ];

    int32_t blkInfoNum = 0;
    auto startIt       = addedHashIds.begin();
    auto curIt         = startIt;

    while (curIt != addedHashIds.end()) {
        if (curIt->parentHashId == startIt->parentHashId) {
            blkInfo[blkInfoNum] = {(uint64_t)curIt->hashId,
                                   (uint64_t)curIt->localHashId};
            LOG_DEBUG_FMT("[Add BlkInfo] hashId:{}, localHashId:{}",
                          curIt->hashId, curIt->localHashId);
            blkInfoNum++;
            curIt++;
            continue;
        }

        publishOneStoredEvent(startIt->parentHashId, 
                            blkInfo, blkInfoNum);
        startIt = curIt;
    }

    if (curIt != startIt) {
        publishOneStoredEvent(startIt->parentHashId, 
                            blkInfo, blkInfoNum);
    }

    return;
}




void DynamoPublisher::publishRemovedEventBatch(
    std::vector<uint64_t> &removedHashIds) {
    if (!removedHashIds.empty()) {
        LOG_DEBUG_FMT("removedHashId:");
        for (auto hashId : removedHashIds) {
            LOG_DEBUG_FMT("[Remove] hashId:{}", hashId);
        }
        LOG_DEBUG_FMT("removedHashId end");
    }

    for (int32_t i = 0; i < removedHashIds.size();
         i += MAX_HASHIDS_PER_NATS_REQ) {
        uint64_t hashIds[MAX_HASHIDS_PER_NATS_REQ];
        int32_t hashIdNum = std::min((int32_t)removedHashIds.size() - i,
                                     MAX_HASHIDS_PER_NATS_REQ);
        LOG_DEBUG_FMT("[Publish Removed Event] hashIdNum:{}", hashIdNum);
        std::copy(removedHashIds.begin() + i,
                  removedHashIds.begin() + i + hashIdNum, hashIds);
        publishRemovedEvent(_eventIdCnt, hashIds, hashIdNum);
        _eventIdCnt++;
    }
    return;
}



/**
 *! @brief 通过新旧两个BlockHashId数组，判断新相对于旧版新增、删除了哪些BlockHashId
 * 
 * @param oldCache       老的BlockHashId数组
 * @param newCache       新的BlockHashId数组
 * @param addedHashIds   新相对于旧版新增了哪些BlockHashId
 * @param removedHashIds 新相对于旧版移除了哪些BlockHashId对应的hashId
 */
void CalcHashIdDelta(std::vector<BlockHashId> &oldCache,
                     std::vector<BlockHashId> &newCache,
                     std::vector<BlockHashId> &addedHashIds,
                     std::vector<uint64_t> &removedHashIds) {
    addedHashIds.clear();
    removedHashIds.clear();

    std::unordered_set<int64_t> oldSet;
    std::unordered_set<int64_t> newSet;

    for (int32_t i = 0; i < oldCache.size(); i++) {
        oldSet.insert(oldCache[i].hashId);
    }

    for (int32_t i = 0; i < newCache.size(); i++) {
        newSet.insert(newCache[i].hashId);
    }

    for (int32_t i = 0; i < oldCache.size(); i++) {
        if (newSet.count(oldCache[i].hashId) == 0) {
            LOG_DEBUG_FMT("[Remove] hashId:{}", oldCache[i].hashId);
            removedHashIds.push_back((uint64_t)oldCache[i].hashId);
        }
    }

    for (int32_t i = 0; i < newCache.size(); i++) {
        if (oldSet.count(newCache[i].hashId) == 0) {
            LOG_DEBUG_FMT("[AddHashId] hashId:{}, parent:{}, local:{}",
                          newCache[i].hashId, newCache[i].parentHashId,
                          newCache[i].localHashId);

            addedHashIds.push_back(newCache[i]);
        }
    }

    return;
}



void DynamoPublisher::reg(MemoryPool *pool) noexcept {
    LOG_DEBUG_FMT("Dynamo Publisher Start Reg!");
    // 检查独立线程是否已启动，不重复启动
    if (_thread.joinable()) {
        LOG_DEBUG_FMT("thread is running!");
        return;
    }


    _thread = std::thread([this, pool]() {
        LOG_DEBUG_FMT("Dynamo Publisher Start!");

        std::vector<uint64_t>    removedHashIds;
        std::vector<BlockHashId> addedHashIds;
        std::vector<BlockHashId> oldHashIds;    // 上次上报时存在的BlockHashId

        while (true) {
            LOG_DEBUG_FMT(
                "Dynamo Publisher waked up! will check and report...");

            // 1.从Memory Pool中获取最新时刻缓存的hashId集合
            auto newHashIds = pool->getAllCachedSegmentIds();

            // 2.将最新缓存的hashId和上一时刻缓存的hashId进行对比，将删除的hashId和新增的hashId分别统计
            CalcHashIdDelta(oldHashIds, 
                            newHashIds, 
                            addedHashIds,
                            removedHashIds
                        );

            // 3.调用Dynamo Event Manager接口，上报删除的hashId
            publishRemovedEventBatch(removedHashIds);

            // 4.调用Dynamo Event Manager接口，上报新增的hashId
            publishStoredEventBatch(addedHashIds);

            oldHashIds = newHashIds;
            sleep(1);
        }
    });

    LOG_DEBUG_FMT("Dynamo Publisher Finish Reg!");

    return;
}



// TODO(yangxianpku: 2025.08.19): 线程是个死循环，直接join有用么？
void DynamoPublisher::unreg() noexcept {
    if (_thread.joinable()) {
        _thread.join();
    }
    return;
}